package com.atguigu;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.util.Collector;
import scala.Tuple2;
import org.apache.flink.configuration.Configuration;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.ArrayList;
import java.util.Random;

public class PhoneOverlapDemo {

    // Generate a list of phone numbers
    private static List<String> generatePhoneNumbers(int num) {
        List<String> phones = new ArrayList<>();
        Random random = new Random();
        for (int i = 0; i < num; i++) {
            phones.add(String.format("%011d", random.nextInt(1000000000)));
        }
        return phones;
    }

    public static void main(String[] args) throws Exception {

        final int[] count = {0};
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // Source 1: customer phone numbers
        DataStream<List<String>> customerPhones = env.addSource(new SourceFunction<List<String>>() {
            @Override
            public void run(SourceContext<List<String>> ctx) throws Exception {
                while (true) {
                    if (count[0] % 10000 == 0) {
                        System.out.println("count = " + count[0]);
                    }
                    ctx.collect(generatePhoneNumbers(1000));
                    Thread.sleep(20);
                    count[0] += 1;
                }
            }

            @Override
            public void cancel() {
            }
        }).returns(Types.LIST(Types.STRING)).name("customerPhones");
//        customerPhones.print();

        // Source 2: fixed group phone numbers
        List<List<String>> fixedGroupPhonesList = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            fixedGroupPhonesList.add(generatePhoneNumbers(1000));
        }
//        System.out.println(fixedGroupPhonesList);
        List<List<List<String>>> fixedGroupPhonesListList = new ArrayList<>();
        fixedGroupPhonesListList.add(fixedGroupPhonesList);
        DataStream<List<List<String>>> fixedGroupPhones = env.fromCollection(fixedGroupPhonesListList).name("fixedGroupPhones");

//        fixedGroupPhones.print();

        // TODO: Calculate phone number overlap and overlap ratio
        // Convert each phone number list to a HashSet
        SingleOutputStreamOperator<HashSet<String>> customerPhonesSet =
                customerPhones.map((MapFunction<List<String>, HashSet<String>>) HashSet::new)
                        .returns(TypeInformation.of(new TypeHint<HashSet<String>>() {
                        })).name("customerPhonesSet");

//        DataStream<HashSet<String>> fixedGroupPhonesSet = fixedGroupPhones.map((MapFunction<List<List<String>>, HashSet<String>>) HashSet::new)
//                .returns(TypeInformation.of(new TypeHint<HashSet<String>>() {
//                }));

        // Connect the two streams
        ConnectedStreams<HashSet<String>, List<List<String>>> connectedStreams = customerPhonesSet.connect(fixedGroupPhones);

        // Compare the two streams and calculate the number of shared phone numbers
        DataStream<Integer> sharedPhoneNumbers = connectedStreams.flatMap(new CoFlatMapFunction<HashSet<String>, List<List<String>>, Integer>() {
            Set<String> customerPhones;
            List<List<String>> fixedGroupPhones;

            @Override
            public void flatMap1(HashSet<String> value, Collector<Integer> out) {
                customerPhones = value;
//                System.out.println("map1 - customerPhones: " + customerPhones.size());
//                if (fixedGroupPhones != null) {
//                    System.out.println("map1 - fixedGroupPhones: " + fixedGroupPhones.size());
//                }
                out.collect(calculateSharedPhoneNumbersRatio(customerPhones, fixedGroupPhones));
            }

            @Override
            public void flatMap2(List<List<String>> value, Collector<Integer> out) {
//                fixedGroupPhones = value;
//                System.out.println("map2 - fixedGroupPhones: " + fixedGroupPhones);
//                return sharedPhoneNumbersHashSet(customerPhones, fixedGroupPhones);
                fixedGroupPhones = value;
                System.out.println("map2 - fixedGroupPhones: " + fixedGroupPhones.size());
            }

//            private int sharedPhoneNumbersHashSet(Set<String> set1, List<List<String>> set2) {
//                System.out.println("sharedPhoneNumbersHashSet - set1: " + set1);
//                System.out.println("sharedPhoneNumbersHashSet - set2: " + set2);
//                if (set1 == null || set2 == null) {
//                    return 0;
//                }
//
//                Set<String> smallerSet, largerSet;
//                if (set1.size() <= set2.size()) {
//                    smallerSet = set1;
//                    largerSet = set2;
//                } else {
//                    smallerSet = set2;
//                    largerSet = set1;
//                }
//
//                int sharedCount = 0;
//                for (String number : smallerSet) {
//                    if (largerSet.contains(number)) {
//                        sharedCount++;
//                    }
//                }
//                return sharedCount;
//            }

            public int calculateSharedPhoneNumbersRatio(Set<String> set1, List<List<String>> list2) {
                if (set1 == null || list2 == null) {
                    return 0;
                }

                long startTime, endTime = 0;
                startTime = System.nanoTime();
                List<Double> ratios = new ArrayList<>();

                for (List<String> list : list2) {
                    int matchingCount = 0;
                    for (String str : list) {
                        if (set1.contains(str)) {
                            matchingCount++;
                        }
                    }

                    double ratio = (double) matchingCount / list.size();
                    ratios.add(ratio);
                }

                int count = 0;
                double threshold = 0.1; // 10%的阈值
                for (double ratio : ratios) {
                    if (ratio > threshold) {
                        count++;
                    }
                }
                endTime = System.nanoTime();
                String threadName = Thread.currentThread().getName();
                System.out.println("Thread："+ threadName + "，" + "超过10%数量：" + count + "，" +"计算耗时：" + (endTime - startTime) + "ns");
                return count;
            }

        });


        // TODO: Print the results and the computation time
//        sharedPhoneNumbers.print();
        /**
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：99737700ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：100364200ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：96694100ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：96579400ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：97818000ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：98703600ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：98848800ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：100002500ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：98939000ns
         * Thread：Co-Flat Map (2/2)#0，超过10%数量：0，计算耗时：97809400ns
         */
        env.execute("PhoneOverlapDemo");
    }
}
